package com.kasoft.emq.runner;

import com.kasoft.emq.handler.Dispatcher;
import lombok.extern.slf4j.Slf4j;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;



@Component
@Slf4j
public class EmqttdRunner implements ApplicationRunner {

	@Value("${emq.host}")
	private String emqHost;

	@Value("${emq.port}")
	private int emqPort;

	@Value("${emq.clientId}")
	private String emqClientId;

	@Override
	public void run(ApplicationArguments args) throws Exception {

		MQTT mqtt=new MQTT();
		mqtt.setHost(emqHost,emqPort);
		mqtt.setClientId(""+ Math.random());
		CallbackConnection connection=mqtt.callbackConnection();

		connection.listener(new Listener() {
			@Override
			public void onConnected() {

			}

			@Override
			public void onDisconnected() {
				log.error("连接丢失");
			}

			@Override
			public void onPublish(UTF8Buffer utf8Buffer, Buffer buffer, Runnable ack) {
				//当有设备向服务已订阅的主题发送消息时,该方法会消费
			//	System.out.println(UTF8Buffer.encode(buffer.toString()));
				// 分发消息给对应的消息处理器
				Dispatcher.submit(utf8Buffer, buffer.ascii(), ack);
				ack.run();
			}

			@Override
			public void onFailure(Throwable throwable) {
				log.error("消费异常:"+ throwable.fillInStackTrace());
			}
		});
		connection.connect(new Callback<Void>() {
			@Override
			public void onSuccess(Void aVoid) {
				//连接成功后会默认订阅主题($runner/mengsu)
				log.info("连接成功");

				//新建一个device主题
				Topic[] topic=new Topic[]{new Topic("Device", QoS.AT_LEAST_ONCE)};
				connection.subscribe(topic, new Callback<byte[]>() {
					@Override
					public void onSuccess(byte[] bytes) {
						log.info("订阅成功");
					}

					@Override
					public void onFailure(Throwable throwable) {

					}
				});

			}
			@Override
			public void onFailure(Throwable throwable) {
				log.error("连接失败");
			}
		});

		connection.getDispatchQueue().execute(new Runnable() {
			@Override
			public void run() {
				// publish
			}
		});
	}



}
